﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Reactive;
using System.Reactive.Subjects;
using System.Text;
using System.Threading.Tasks;

namespace System.Threading.Actors.Utilities
{
    // Common logic for handling pub/sub functionality within an IActorState implementation.
    public class PubSubHelper
    {
        private IActorState m_state;
        private String m_partitionName;

        // Maintains a Subject<string> for each event type/topic
        protected readonly Dictionary<string, ISubject<ActorEvent>> subjects = new Dictionary<string, ISubject<ActorEvent>>();

        // Maintains a list of subscriptions for each client.  A client can subscribe to multiple event types, so we need to track a subscription per event type per client.
        protected readonly Dictionary<IPublicationCallbackContract, List<Tuple<string, IDisposable>>> subscriptions = new Dictionary<IPublicationCallbackContract, List<Tuple<string, IDisposable>>>();

        private bool subscribed = false; // Set to true if there are ANY subscriptions

        // Events are published in the background to minimize the risk of deadlock.
        // But running them through the m_publicationQueue ensures that they will
        // be published in the order in which they arrive.
        private AsyncQueue<ActorEvent> m_publicationQueue = new AsyncQueue<ActorEvent>();

        private ActorETWEventListener m_etwListener;

        public PubSubHelper(IActorState state, String partitionName, ActorETWEventListener listener = null)
        {
            m_state = state;
            m_partitionName = partitionName;
            m_etwListener = listener;
            if (listener != null)
            {
                subjects["ETW"] = m_etwListener.ActorEventSubject;
            }
            PublicationTaskLoop(); // Fire-and-forget
        }

        // Allows a client to subscribe to an event type
        public void Subscribe(IPublicationCallbackContract callerProxy, string eventType)
        {
            Contract.Requires(!String.IsNullOrWhiteSpace(eventType), "Event type can't be null or whitespace");
            Contract.Requires(Contract.ForAll(ActorClient.ReservedDelimiters, reservedChar => !eventType.Contains(reservedChar)), "Actor event names cannot use '$'.  That character is reserved.");

            // If necessary, add new type to subject dictionary
            lock (subjects)
            {
                if (!subjects.ContainsKey(eventType))
                {
                    subjects[eventType] = new Subject<ActorEvent>();
                }
            }

            // Subscribe to appropriate subject
            var subscription = subjects[eventType].Subscribe(new CallbackObserver(callerProxy, m_state, m_partitionName));

            // Store subscription information in subscriptions dictionary
            lock (subscriptions)
            {
                if (!subscriptions.ContainsKey(callerProxy))
                {
                    subscriptions.Add(callerProxy, new List<Tuple<string, IDisposable>>());
                }
                this.subscriptions[callerProxy].Add(Tuple.Create(eventType, subscription));
            }

            subscribed = true;
            //Console.WriteLine("({0}): Received subscription for {1}", this.GetHashCode(), eventType);
        }

        // Allow a client to unsubscribe from an event type
        public void Unsubscribe(IPublicationCallbackContract callerProxy, string eventType)
        {
            string status = null;
            // Lock down the subscriptions dictionary while we peruse it
            lock (subscriptions)
            {
                List<Tuple<string, IDisposable>> subscriptionTypeList;
                if (subscriptions.TryGetValue(callerProxy, out subscriptionTypeList))
                {
                    // We found the type list for our client.  Now remove the appropriate element/subscription from the type list.
                    var theRightSubscription = subscriptionTypeList.Where(sub => sub.Item1 == eventType).FirstOrDefault();
                    if (theRightSubscription != null)
                    {
                        theRightSubscription.Item2.Dispose();
                        subscriptionTypeList.Remove(theRightSubscription);
                        status = String.Format("Unsubscribe({0}) successful", eventType);
                    }
                    else
                    {
                        status = String.Format("Unsubscribe({0}) FAILED to find correct subscription", eventType);
                    }
                }
                else
                {
                    status = String.Format("Unsubscribe({0}) FAILED to find subscription list", eventType);
                }
            }

            //Console.WriteLine(status);
        }

        // Allow a client to unsubscribe from all events
        public void UnsubscribeAll(IPublicationCallbackContract callerProxy)
        {
            // Lock down the subscriptions list while we tinker with it
            lock (subscriptions)
            {
                if (subscriptions.ContainsKey(callerProxy))
                {
                    // We've got the type list for our client.  Now unsubscribe from all of them and remove the list from the subscriptions dictionary.
                    var subscriptionTypeList = subscriptions[callerProxy];
                    foreach (var subscriptionType in subscriptionTypeList)
                    {
                        subscriptionType.Item2.Dispose();
                    }
                    subscriptions.Remove(callerProxy);
                }
            }
        }

        public void UnsubscribeEverything()
        {
            // Lock down the subscriptions list while we tinker with it
            lock (subscriptions)
            {
                try
                {
                    foreach (var subscription in subscriptions.Values)
                    {
                        var subscriptionTypeList = subscription;
                        foreach (var subscriptionType in subscriptionTypeList)
                        {
                            subscriptionType.Item2.Dispose();
                        }
                    }
                }
                finally
                {
                    subscriptions.Clear();
                }
            }
        }

        // Publish to an eventType with an eventValue
        public void Publish(string eventType, object eventPayload, params object[] eventPayloadQualifiers)
        {
            if (!subscribed) return; // Nobody has subscribed to this actor, so no work to do.

            var newEvent = new ActorEvent(eventType, eventPayload, eventPayloadQualifiers);
            var serviceName = m_state.GetId();
            newEvent.EventSource = serviceName;
            newEvent.EventSourcePartitionName = m_state.PartitionName;
            // TODO: I really don't like forcing Enqueue to spin up a thread in order to wake up a waiting task.  That's
            // going to be slow.  In the future, maybe we should batch
            m_publicationQueue.Enqueue(newEvent, underLock: false);
            ActorETWEventSource.Log.EventPublished(serviceName, newEvent); // TODO: Do we really need to pass serviceName, as it's encoded in newEvent?
        }

        public void Publish(string eventType, object eventPayload)
        {
            Publish(eventType, eventPayload, null);
        }

        // Async loop that waits for publications and processes them
        private async void PublicationTaskLoop()
        {
            while (true)
            {
                ActorEvent actorEvent;
                try
                {
                    actorEvent = await m_publicationQueue.DequeueAsync().ConfigureAwait(continueOnCapturedContext:false); // this will resume when an event is available
                }
                catch (EndOfQueueException)
                {
                    // The queue is empty and marked as "completed"
                    return;
                }

                var eventType = actorEvent.EventType;

                ISubject<ActorEvent> subject;

                try
                {
                    if (subjects.TryGetValue(eventType, out subject))
                        subject.OnNext(actorEvent);
                }
                catch (Exception e)
                {
                    // @TODO: Log an event?  Or could that start an infinite loop, if our ETW logging fails?
                    Console.WriteLine("Event processing for event type {0} threw an exception: {1}", eventType, e.Message);
                    Trace.WriteLine(String.Format("Event processing for event type {0} threw an exception: {1}", eventType, e.Message));
                }

                try
                {
                    // Wildcard support
                    // TODO: Users can subscribe to both a wildcard and a specific event, then get two notifications for the event.
                    // Clean that up when we do a proper wildcard/regex based solution.
                    if (subjects.TryGetValue("*", out subject))
                        subject.OnNext(actorEvent);
                }
                catch (Exception e)
                {
                    Console.WriteLine("Event processing for all events threw an exception: {0}", e.Message);
                    Trace.WriteLine("Event processing for all events threw an exception: {0}", e.Message);
                }
            }
        }

        public void Close()
        {
            m_publicationQueue.CompleteAdding();
        }

        // Utility class that stores subscription information into an observer,
        // to be called later during event publication.
        class CallbackObserver : ObserverBase<ActorEvent>
        {
            private readonly IPublicationCallbackContract _callerProxy;  // Stores the proxy to the caller/callback
            private readonly IActorState _state;
            private readonly String _partitionName;

            public CallbackObserver(IPublicationCallbackContract callerProxy, IActorState state, String partitionName)
            {
                _callerProxy = callerProxy;
                _state = state;
                _partitionName = String.IsNullOrEmpty(partitionName) ? null : partitionName;
            }

            protected override void OnNextCore(ActorEvent value)
            {
                try
                {
                    if ((_partitionName == value.EventSourcePartitionName) ||
                        (_partitionName == null && String.IsNullOrEmpty(value.EventSourcePartitionName)))
                        _callerProxy.OnNewEvent(value);
                }
                catch (Exception)
                {
                    Trace.WriteLine("Dead subscriber detected; un-subscribing it.");
                    _state.Unsubscribe(_callerProxy, value.EventType);
                }
            }

            protected override void OnCompletedCore()
            {
                try
                {
                    _callerProxy.OnCompleting();
                }
                catch (Exception)
                {
                    //TODO: Log this subscriber as dead
                }
            }

            protected override void OnErrorCore(Exception error)
            {
                Trace.WriteLine("Saw OnErrorCore in CallbackObserver; exc = {0}", error.ToString());
                Console.WriteLine("Saw OnErrorCore in CallbackObserver; exc = {0}", error.ToString());
                // Do nothing for now
            }
        }
    }

}
